package com.hexiang.rocketmq.consumer;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.annotation.SelectorType;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;

import java.util.concurrent.TimeUnit;

@Slf4j
@Component
@RocketMQMessageListener(topic = "test_topic", consumerGroup = "mock-reconsume-by-timeout-spring-consumer-group", selectorExpression = "*", selectorType = SelectorType.TAG, consumeTimeout = 1)
public class MockReConsumeByTimeOutRocketMQListener implements RocketMQListener<MessageExt> {

    @Override
    public void onMessage(MessageExt messageExt) {

        //逐条消费
        String messageBody = new String(messageExt.getBody());
        log.info("Message Consumer: Handle New Message: messageId: " + messageExt.getMsgId() + ",topic: " +
                messageExt.getTopic() + ",tags: " + messageExt.getTags() + ",messageBody: " + messageBody);

        //模拟耗时操作2分钟，大于设置的消费超时时间1分钟
        try {
            TimeUnit.MINUTES.sleep(2);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }

    }

}